/**
 * Created by zhangguangde on 2017/4/28.
 */
var util = require('util');
var events = require('events');
var fs = require('fs');
var redis = require("redis");
    client = redis.createClient();


var totalCount = {};

var inputFile = './json/20170309.json';

function ReadStreamThrottle (stream, speed) {
    this._stream = stream;
    this._readBytes = 0;
    this._speed = speed;
    this._ended = false;
    this._readBytesSecond = 0;
    this._lastTimestamp = Date.now();
    this._paused = false;
    var self = this;


    var lastLine = "";

    // test用
    var k = 0;

    // 每次取出64kb
    stream.on('data', function (chunk) {


        self._readBytes += chunk.length;
        self._readBytesSecond += chunk.length;
        self.emit('data', chunk);


        // 暂停
        self.pause();

        // buffer转换string
        chunk = chunk.toString();
        var lines = chunk.split('\n');

        try{

            lines.forEach(function (line, index) {

                var json = JSON.parse(lastLine + line);
                // 清空上次取出的最后半条
                lastLine = "";

                var param = {
                    time:json.createAt.s,
                    packageName: json.currentPackageName.s,
                    country: json.country.s,
                    city: json.city.s,
                    count: 1
                };

                // 判断对象中是否存在，存在的场合，计数字段加 1，不存在的场合，添加key
                if (totalCount[json.createAt.s + '_' + json.currentPackageName.s + '_' + json.country.s + '_' + json.city.s]) {
                    // 存在的场合，计数字段加 1
                    totalCount[json.createAt.s + '_' + json.currentPackageName.s + '_' + json.country.s + '_' + json.city.s].count++;
                } else {
                    // 不存在的场合，添加key
                    totalCount[json.createAt.s + '_' + json.currentPackageName.s + '_' + json.country.s + '_' + json.city.s] = param;
                }

                //console.log(param);
            });

        }
        // 捕获最后未满一行的异常
        catch(e){
            //console.log(lines[lines.length - 1]);
            lastLine = lines[lines.length - 1];
        }

        //console.log(totalCount);

        // 操作结束后继续执行读取
        self.resume();
        //if (k < 1) {
        //    self.resume();
        //    k++;
        //}

    });
    // 监听读取结束动作
    stream.on('end', function () {
        self._ended = true;
        self.emit('end');

        var buffer = new Buffer(JSON.stringify(totalCount));

        console.log('占用字节数：',  buffer.length);
        var y = 1;
        for ( var key in totalCount ) {
            //console.log(y++);
            //console.log(totalCount[key]);
            // 存储到redis
            client.HMSET(key, totalCount[key], function(err, result) {
                if (err) {
                    //console.log('存储失败');
                } else {
                    //console.log('存储成功');
                }
            });
        }

    });
}

util.inherits(ReadStreamThrottle, events.EventEmitter);

ReadStreamThrottle.prototype.pause = function () {
    this._paused = true;
    this._stream.pause();
};

ReadStreamThrottle.prototype.resume = function () {
    this._paused = false;
    this._stream.resume();
};


// 限制速度
var MB = 1024 * 1024;
var s = new ReadStreamThrottle(fs.createReadStream(inputFile), MB * 1);
var bytes = 0;
var t = Date.now();
s.on('data', function (c) {
    //console.log(c.length);
    bytes += c.length;
    var spent = (Date.now() - t) / 1000;
    //console.log('read %s bytes, speed: %sMB/S', bytes, (bytes / MB / spent).toFixed(2));
});
s.on('end', function () {
    console.log('end. total %s bytes', bytes);
});



// 打印内存详情
//function printMemoryUsage () {
//    var info = process.memoryUsage();
//    function mb (v) {
//        return (v / 1024 / 1024).toFixed(2) + 'MB';
//    }
//    console.log('rss=%s, heapTotal=%s, heapUsed=%s', mb(info.rss), mb(info.heapTotal), mb(info.heapUsed));
//}
//setInterval(printMemoryUsage, 1000);